package com.atguigu.dga.governance.assessor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.common.util.SqlParse;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.google.common.collect.Sets;
import lombok.Getter;
import lombok.Setter;
import org.antlr.runtime.tree.Tree;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;


@Component("IS_SIMPLE_PROCESS")
public class IsSimpleProcessAssessor extends Assessor {

    //1    调用语法树解析工具
    //2    要制作一个节点处理器：
    //                2.1   收集语句中的：任何join\groupby\\union\函数的节点
    //                 2.2   收集来源表的表名
    //                 2.3    收集where 后面的条件字段
    //3     定性 ：3.1  如果包含任何复杂计算（ join\groupby\\union\ ） =>非简单加工
    //                   3.2 如果不包含复杂计算
    //                             根据来源表名提取来源表的元数据，分区字段
    //                              对比条件字段是否只包含分区字段  如果是  ==》简单加工
    //4      评分写批语
    @Override
    public void checkProblem(AssessParam assessParam, GovernanceAssessDetail governanceAssessDetail) throws Exception {
        if(assessParam.getTDsTaskDefinition()==null) {
            return;
        }

        String sql = assessParam.getTDsTaskDefinition().getSql();
         if(sql==null){
             return;
         }
        //1    调用语法树解析工具
        IsSimpleProcessDispatcher dispatcher= new IsSimpleProcessDispatcher();
        dispatcher.setDefaultSchemaName(assessParam.getTableMetaInfo().getSchemaName());


        //2 采集阶段
        SqlParse.parse(sql,dispatcher);
        Set<String> curSqlComplicationProcessSet = dispatcher.getCurSqlComplicationProcessSet();
        Set<String> sourceTableSet = dispatcher.getSourceTableSet();
        Set<String> whereColNameSet = dispatcher.getColNameSet();

        System.out.println("colNameSet = " + whereColNameSet);

        //3     定性 ：3.1  如果包含任何复杂计算（ join\groupby\\union\ ） =>非简单加工
        //                   3.2 如果不包含复杂计算
        //                             根据来源表名提取来源表的元数据，分区字段
        //                              对比条件字段是否只包含分区字段  如果是  ==》简单加工
        if(curSqlComplicationProcessSet.size()>0){
            governanceAssessDetail.setAssessComment("包含复杂计算，非简单加工:"+ StringUtils.join(curSqlComplicationProcessSet, ",") );

        }else{
            Map<String, TableMetaInfo> tableMetaInfoMap = assessParam.getTableMetaInfoMap();

            for (String sourceTableFullName  : sourceTableSet) {
                TableMetaInfo tableMetaInfo = tableMetaInfoMap.get(sourceTableFullName);//获得来源表的元数据
                //把分区字段json加工为 一个只有分区字段名的set
                String partitionColNameJson = tableMetaInfo.getPartitionColNameJson();
                List<JSONObject> partitionColJSONobjList = JSON.parseArray(partitionColNameJson, JSONObject.class);
                Set<String> partitionColNameSet = partitionColJSONobjList.stream().map(jsonObject -> jsonObject.getString("name")).collect(Collectors.toSet());
                whereColNameSet.removeAll(partitionColNameSet); // 去除where条件中的分区字段  ，剩下的就是非分区字段

            }

            if(whereColNameSet.size()==0){  //没有非分区字段
                 governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
                 governanceAssessDetail.setAssessProblem("没有复杂计算，且过滤条件皆为分区字段");
            }
            governanceAssessDetail.setAssessComment("包含非分区字段："+StringUtils.join(whereColNameSet, ","));

        }



    }


    class  IsSimpleProcessDispatcher implements Dispatcher{


        //                2.1   收集语句中的：任何join\groupby\\union\函数的节点
        //                 2.2   收集来源表的表名
        //                 2.3    收集where 后面的条件字段

        Set<Integer>  complicationPrcessSet= Sets.newHashSet(HiveParser.TOK_JOIN,
                                                    HiveParser.TOK_FULLOUTERJOIN,
                                                    HiveParser.TOK_LEFTOUTERJOIN,
                                                    HiveParser.TOK_RIGHTOUTERJOIN,
                                                    HiveParser.TOK_UNIONALL,
                                                    HiveParser.TOK_FUNCTION,
                                                    HiveParser.TOK_FUNCTIONSTAR,
                                                    HiveParser.TOK_FUNCTIONDI,
                                                    HiveParser.TOK_SUBQUERY,
                                                    HiveParser.TOK_SELECTDI,
                                                    HiveParser.TOK_GROUPBY
                );

        // 收集复杂操作
        @Getter
        Set<String>  curSqlComplicationProcessSet= new HashSet<>();
        //收集来源表
        @Getter
        Set<String>  sourceTableSet= new HashSet<>();
        //收集where 后面的条件字段
        @Getter
        Set<String>  colNameSet= new HashSet<>();
        @Setter
        String defaultSchemaName =null;


        @Override
        public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs) throws SemanticException {
           //  2.1   收集语句中的：任何join\groupby\\union\函数的节点
            ASTNode astNode = (ASTNode) nd;
           if(  complicationPrcessSet.contains(astNode.getToken().getType())  ){
               curSqlComplicationProcessSet.add(astNode.getText());
           }
            //    2.2   收集来源表的表名
                  //锚点 ：  TOK_TABREF
            if(astNode.getToken().getType()==HiveParser.TOK_TABREF){
                ASTNode tokTableNameNode = (ASTNode)astNode.getChild(0);
                if(tokTableNameNode.getChildCount()==2){  //带库名
                    String schemaName = tokTableNameNode.getChild(0).getText();
                    String tableName = tokTableNameNode.getChild(1).getText();
                    String fullTableName = schemaName+"."+tableName;
                    sourceTableSet.add(fullTableName);
                }else if(tokTableNameNode.getChildCount()==1){  //不带库名
                    String schemaName = defaultSchemaName;
                    String tableName = tokTableNameNode.getChild(0).getText();
                    String fullTableName = schemaName+"."+tableName;
                    sourceTableSet.add(fullTableName);
                }

            }
            //   收集where 后面的条件字段
            if( astNode.getAncestor(HiveParser.TOK_WHERE) !=null   ){
                if(astNode.getToken().getType()==HiveParser.TOK_TABLE_OR_COL){
                    // 有表名字段
                    if(astNode.getParent().getType()==HiveParser.DOT) { //父级为点，子级一定是表名，取兄弟节点
                        String colName = astNode.getParent().getChild(1).getText();
                        colNameSet.add(colName);
                    }else{//父级不为点，子级一定是字段名
                        String colName = astNode.getChild(0).getText();
                        colNameSet.add(colName);
                    }
                }
            }


            return null;
        }
    }
}
